package org.databandtech.flink.sink;

import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class SinkToMySQL extends RichSinkFunction<Tuple3<String,Integer,BigDecimal>> {

	Connection connection = null;
    PreparedStatement preparedStatement = null;
    String[] COLUMNS;
    Tuple item;
    String SQL;
	String URL = "";
    String USERNAME = "";
    String PASSWORD = "";
    public SinkToMySQL(String uRL, String uSERNAME, String pASSWORD,String sQL) {
		super();
		SQL = sQL;
		URL = uRL;
		USERNAME = uSERNAME;
		PASSWORD = pASSWORD;
	}

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open( parameters );
        String driver = "com.mysql.jdbc.Driver";
        Class.forName( driver );
        connection = DriverManager.getConnection( URL, USERNAME, PASSWORD );
        preparedStatement = connection.prepareStatement(SQL);
    }

    @Override
    public void close() throws Exception {
        super.close();
        if (connection != null) {
            connection.close();
        }
        if (preparedStatement != null) {
            preparedStatement.close();
        }
    }

    @Override
    public void invoke(Tuple3<String,Integer,BigDecimal> tuple, Context context) throws Exception {
        try {
        	//System.out.println(tuple.f0);
        	if (tuple.f0!=null)
            preparedStatement.setString( 1, tuple.f0 );
        	if (tuple.f1!=null)
            preparedStatement.setInt( 2, tuple.f1 );
        	if (tuple.f2!=null)
            preparedStatement.setBigDecimal( 3, tuple.f2 );
            preparedStatement.executeUpdate();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
